require_relative 'kafka_client'

module ReadAndInput
  READ_SIZE = 100
  NUMS_SIZE = 10

  def self.run
    f = File.open(datafile, 'r')
    nums, buf, pos = [], '', 0

    while true
      pos = f.pos
      buf_tmp = f.read(READ_SIZE)
      break if buf_tmp.nil?

      buf_arr = (buf + buf_tmp).split("\n")
      buf = buf_arr.pop

      nums.concat buf_arr
      if nums.size > NUMS_SIZE
        KafkaClient.send_message nums
        nums = []
      end
    end

    if nums.size > 0
      KafkaClient.send_message nums
      nums = []
    end
  rescue => e
    puts e.message
    puts e.backtrace
  ensure
    puts "pos: [#{pos}]"
    f.close
  end

  def self.datafile
    __FILE__.sub /[^ \/]+$/, '../tmp/data.txt'
  end
end
